/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog.pipe;

import java.io.IOException;
import java.lang.reflect.UndeclaredThrowableException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.changelog.ChangeLogServer;
import com.nhncorp.neptune.changelog.pipe.Bulk.OperationResult;
import com.nhncorp.neptune.changelog.pipe.Pipe.Context;
import com.nhncorp.neptune.common.testhelper.FaultInjectionProxy;
import com.nhncorp.neptune.common.testhelper.ProxyExceptionHelper;

class RunState extends PipeState implements RunStateTestIF {
  static final Log LOG = LogFactory.getLog(RunState.class);
  private static final PipeState singleton = new RunState();
  
  public static PipeState instance() {
    return singleton;
  }
  
  RunStateTestIF testProxy;
  RunState() {
    testProxy = this;
    if (ChangeLogServer.IS_TEST_MODE) {
      testProxy = FaultInjectionProxy.wrap(this, RunStateTestIF.class);
    }
    
//    faThread = new FileAppendThread();
//    faThread.setDaemon(true);
//    faThread.start();
  }

  public void init(Pipe.Context ctx) throws IOException {
//    LOG.debug("RUN STATE!!!");
    ctx.registerToSelect(ctx.prevChannel, SelectionKey.OP_READ);
    ctx.prevChannel.socket().setTcpNoDelay(true);
    ctx.prevChannel.configureBlocking(false);

    if (ctx.nextChannel != null) {
      ctx.nextChannel.configureBlocking(false);
      ctx.nextChannel.socket().setSendBufferSize(21900);
      ctx.nextChannel.socket().setReceiveBufferSize(43800);
      ctx.nextChannel.socket().setTcpNoDelay(true);
      ctx.registerToSelect(ctx.nextChannel, SelectionKey.OP_READ);
    }
    ctx.bulk = new Bulk();
  }
  
  public void close(Context ctx) throws IOException {
    ctx.deregisterFromSelectAll(ctx.prevChannel);
    if (ctx.nextChannel != null) {
      ctx.deregisterFromSelectAll(ctx.nextChannel);
    }
    ctx.bulk = null;
  }

  public boolean fini(Pipe.Context ctx, int bytes) throws IOException {
    /* 
     * by sangchul
     * 사용하지 않는 기능이다.
     */
    
    /*
    ctx.waitForReadingDone(bytes);
    testProxy.writeToLogFile(ctx);
    ctx.checkValidity.set(false);
    ctx.inReceivingData.set(false);
    */
    return true;
  }

  long before;
  long after;

  public boolean readFromNext(Pipe.Context ctx) throws IOException {
    ctx.heartBeatBuf.clear();
    int numRead = ctx.nextChannel.read(ctx.heartBeatBuf);

    if (numRead < 0) {
      throw new IOException("Disconnect from next");
    }
    
    if (numRead >= Pipe.HEART_BEAT_SIZE) {
      ctx.heartBeatBuf.position(0);
      writeToPrev(ctx);
    } else {
      LOG.info("reading heartbeat msg is pending in next select time");
    }

    return true;
  }

  public boolean readFromPrev(Pipe.Context ctx) throws IOException {
    OperationResult ret = ctx.bulk.read(ctx.prevChannel);

    if (ctx.inReceivingData.getAndSet(true) == false) {
      // file channel의 rollback 위치는 fini를 수신해 파일에 기록을 한 후
      // 다음 번 데이터가 들어올 때 다시 갱신된다.
      // 왜냐하면, 다음 번 데이터가 들어온다는 이야기는 commit이 성공적으로 끝났다는 걸 의미하며,
      // 이는 다시 이 위치로 rollback할 일이 없다는 걸 의미한다.
      //LOG.debug("reset rollback index in read");
      //ctx.fileChannel.resetRollbackIndex();

      // pipe의 validity를 체크해서 timeout동안 데이터가 수신되지 않으면
      // 끊어버린다.
      ctx.checkValidity.set(true);
    }

    if (ret == OperationResult.completed) {
      ctx.inReceivingData.set(false);
    }
    
    writeToNext(ctx);

    return true;
  }

  public boolean writeToNext(Pipe.Context ctx) throws IOException {
    if (ctx.isLastPipe && ctx.inReceivingData.get() == false) {
//      LOG.debug("READ COMPLETED");
      
      try {
        testProxy.writeToLogFile(ctx);
      } catch(UndeclaredThrowableException e) {
        ProxyExceptionHelper.handleException(e, LOG);
      } finally {
        ctx.bulk.clear();
      }
      
      ctx.inReturning.set(true);

      ctx.heartBeatBuf.clear();
      ctx.heartBeatBuf.put((byte) 0xff);
      ctx.heartBeatBuf.flip();
      writeToPrev(ctx);
      
      return true;
    }

    if (ctx.nextChannel != null) {
      if (ctx.bulk.write(ctx.nextChannel) == OperationResult.partially) {
//        LOG.debug("writeToNext is done with partially writing data. selecting next writing event");
        ctx.registerToSelect(ctx.nextChannel, SelectionKey.OP_WRITE);
        ctx.selectingWrite = true;
        return false;
      } else if (ctx.inReceivingData.get() == false && ctx.bulk.isWrittingDone()) {
//        LOG.debug("READ & WRITE COMPLETED");
        try {
          testProxy.writeToLogFile(ctx);
        } catch (UndeclaredThrowableException e) {
          ProxyExceptionHelper.handleException(e, LOG);
        } finally {
          ctx.bulk.clear();
        }
      }

      if (ctx.selectingWrite) {
        ctx.deregisterFromSelect(ctx.nextChannel, SelectionKey.OP_WRITE);
        ctx.selectingWrite = false;
      }
    }

    return true;
  }

  // by sangchul
  // 이 메소드는 pipe를 다루는 쓰레드와 다른 쓰레드가 들어온다.
  // 하지만, 그 다른 쓰레드는 pipe에 이미 정해진 수의 바이트가 들어오기 전까진 block하고 있다가
  // 그 후에 이 메소드를 호출하기 때문에 bufferList가 여러 쓰레드에 의해 공유되는 상황은 발생하지 않는다.
  public void writeToLogFile(Pipe.Context ctx) throws IOException {
    ByteBuffer[] bufferArray = ctx.bulk.getBufferArray();

    try {
      ctx.fileChannel.write(bufferArray);
    } catch (IOException e) {
      LOG.warn("writing content to file is fail", e);
//      ctx.fileChannel.rollback();
      throw e;
    }
//    faThread.register(ctx.fileChannel, ctx.bulk.getBufferArray());
  }

  public boolean writeToPrev(Pipe.Context ctx) throws IOException {
    int numWritten = ctx.prevChannel.write(ctx.heartBeatBuf);

    if (numWritten < Pipe.HEART_BEAT_SIZE) {
      ctx.registerToSelect(ctx.prevChannel, SelectionKey.OP_WRITE);
      LOG.info("heartbeat will be sent in next select time!!");
    } else {
      ctx.heartBeatBuf.clear();
      ctx.deregisterFromSelect(ctx.prevChannel, SelectionKey.OP_WRITE);
      
      ctx.inReturning.set(false);
      ctx.checkValidity.set(false);
    }
    
    return true;
  }

  public String getStateName() {
    return "Run State";
  }
}
